package net.bwie.realtime.dy.dwd.log.job;


import com.alibaba.fastjson.JSON;

import net.bw.realtime.dy.common.utils.*;
import net.bwie.realtime.dy.dwd.log.bean.*;
import net.bwie.realtime.dy.dwd.log.funcation.ql.*;

import net.bwie.realtime.dy.dwd.log.funcation.zl.interractionfuncation.RealTimeInteractionOptimized;
import net.bwie.realtime.dy.dwd.log.funcation.zl.messagealterfuncation.MessageAlertProcessFunction;
import net.bwie.realtime.dy.dwd.log.funcation.zl.messagealterfuncation.MessageCountAggregator;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


import java.time.Duration;
import java.util.List;
import java.util.Map;


/**
 * @ClassName：KafkaDataEtlJob
 * @Author: Vvn
 * @Date: 2025/5/27 9:28
 * @Description: 必须描述类做什么事情, 实现什么功能
 */

public class KafkaDataEtlJob {


    public static void main(String[] args) throws Exception {

        // 执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // DataStreamApi 读取  kafka数据
        DataStream<String> kafkaDataStream = KafkaUtil.consumerKafka(env, "live_events");

        // DataStream etl 过滤脏数据
        DataStream<String> etlDataStream = etlNoJsonData(kafkaDataStream);
        // 转换为实体类
        SingleOutputStreamOperator<TikTokBean> tiktokBeanStream = etlDataStream.process(new ProcessFunction<String, TikTokBean>() {

            @Override
            public void processElement(String value,
                                       ProcessFunction<String, TikTokBean>.Context ctx,
                                       Collector<TikTokBean> out) throws Exception {

                TikTokBean tikTokBean = JSON.parseObject(value, TikTokBean.class);
                out.collect(tikTokBean);
            }
        });

        // 设置水位线
        SingleOutputStreamOperator<TikTokBean> watermarks = tiktokBeanStream
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TikTokBean>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TikTokBean>() {
                            @Override
                            public long extractTimestamp(TikTokBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }));

        // 根据直播间id分组
        KeyedStream<TikTokBean, String> KeyByStream = watermarks.keyBy(new KeySelector<TikTokBean, String>() {
            @Override
            public String getKey(TikTokBean value) throws Exception {
                return value.getRoomId();
            }
        });

        // 开窗
        WindowedStream<TikTokBean, String, TimeWindow> stream =
                KeyByStream.window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)));

        // 开窗聚合 实时求取在线人数--------------------------------------------------------------------------------------------------------
        DataStream<UserCountBean> UserCountStream = stream.apply(new realTimeUserCountFunction());
        DataStream<String> UserCountStringStream = UserCountStream.map(new MapFunction<UserCountBean, String>() {
            @Override
            public String map(UserCountBean value) throws Exception {
                return value.getWindowStartTime() +
                        "," + value.getWindowEndTime() +
                        "," + value.getRoomId() + ","
                        + value.getUserCount() + "," + value.getTs();
            }
        });
       // UserCountStringStream.print();
        JdbcUtil.sinkToClickhouseUpsert(UserCountStringStream, "insert into real_time_tiktok_report.tiktok_user_count values(?,?,?,?,?)");
        // 开窗聚合 实时求取礼物打赏----------------------------------------------------------------------------------------------------------
        DataStream<RoomGiftMoney> RoomGiftMoneyStream = stream.apply(new realTimeGiftFunction());
        DataStream<String> RoomGiftMoneyStringStream = RoomGiftMoneyStream.map(new MapFunction<RoomGiftMoney, String>() {
            @Override
            public String map(RoomGiftMoney value) throws Exception {
                return value.getWindowStartTime() + "," + value.getWindowEndTime() + "," + value.getRoomId() + "," + value.getGiftMoney() + "," + value.getTs();
            }
        });
       // RoomGiftMoneyStringStream.print();

        JdbcUtil.sinkToClickhouseUpsert(RoomGiftMoneyStringStream, "insert into real_time_tiktok_report.tiktok_gift_money values(?,?,?,?,?)");

        // 互动消息热度分析----------------------------------------------------------------------------------------------------------
        // 以直播间为主题 求取不同类型的 互动消息 如 share like message
        SingleOutputStreamOperator<RoomInteraction> interactionDataStream = stream.aggregate(new RealTimeInteractionOptimized.InteractionAggregator(), new RealTimeInteractionOptimized.InteractionWindowProcessor());
        DataStream<String> InteractionStringStream = interactionDataStream.map(new MapFunction<RoomInteraction, String>() {
            @Override
            public String map(RoomInteraction value) throws Exception {
                return value.getWindowStartTime() + "," + value.getWindowEndTime() +
                        "," + value.getRoomId() + "," + value.getInteractionCount() + ","
                        + value.getShareCount() + "," + value.getLikeCount() + "," +
                        value.getMessageCount() + "," + value.getTs();
            }
        });
        JdbcUtil.sinkToClickhouseUpsert(InteractionStringStream, "insert into real_time_tiktok_report.tiktok_interaction values(?,?,?,?,?,?,?,?)");
        // 刷礼物异常检测
        // 检测同一用户在 10秒内 累计送礼价值超过 10万元，通过 滚动窗口 聚合实现
//        WindowedStream<TikTokBean, String, TimeWindow> window =
//                watermarks.filter(event -> "gift".equals(event.getEventType()))
//                        .keyBy(TikTokBean::getUserId)
//                        .window(TumblingEventTimeWindows.of(Time.seconds(10)));
//
//        DataStream<AlertGiftSender> AlertGiftStream = window.apply(new realTimeGiftAlertFunction());
//        SingleOutputStreamOperator<AlertGiftSender> aggregate = window.aggregate(new GiftAmountAggregator(), new GiftAlertProcessFunction());
//        DataStream<String> AlertGiftStringStream = aggregate.map(new MapFunction<AlertGiftSender, String>() {
//            @Override
//            public String map(AlertGiftSender value) throws Exception {
//                return value.getUserId() + "," + value.getWindowStartTime() + "," + value.getWindowEndTime() + "," + value.getSendMoney() + "," + value.getTs();
//            }
//        });
//       // AlertGiftStringStream.print();
//
//        JdbcUtil.sinkToClickhouseUpsert(AlertGiftStringStream, "insert into real_time_tiktok_report.tiktok_alert_gift values(?,?,?,?,?)");


        // 刷礼物异常检测：使用 CEP 检测连续小金额送礼行为
        Pattern<TikTokBean, ?> giftAlertPattern = Pattern.<TikTokBean>begin("first-gift")
                .where(new SimpleCondition<TikTokBean>() {
                    @Override
                    public boolean filter(TikTokBean event) {
                        return "gift".equals(event.getEventType()) && event.getGiftValue() <= 1.0;
                    }
                })
                .timesOrMore(100)
                .greedy()
                .within(Time.seconds(10));

// 将模式应用到流
        PatternStream<TikTokBean> patternStream = CEP.pattern(watermarks.keyBy(TikTokBean::getUserId), giftAlertPattern);

// 提取匹配到的异常送礼行为
        SingleOutputStreamOperator<AlertGiftSender> alertStream = patternStream.select(new PatternSelectFunction<TikTokBean, AlertGiftSender>() {
            @Override
            public AlertGiftSender select(Map<String, List<TikTokBean>> patternEvents) {
                List<TikTokBean> events = patternEvents.get("first-gift");
                if (events != null && !events.isEmpty()) {
                    TikTokBean first = events.get(0);
                    long windowStart = events.get(0).getTs();
                    long windowEnd = windowStart + 10000; // 10秒窗口
                    double totalGiftValue = events.stream().mapToDouble(TikTokBean::getGiftValue).sum();

                    return new AlertGiftSender(
                            first.getUserId(),
                            DateTimeUtil.convertLongToString(windowStart,"yyyy-MM-dd HH:mm:ss:SS"),
                            DateTimeUtil.convertLongToString(windowEnd,"yyyy-MM-dd HH:mm:ss:SS"),
                            totalGiftValue,
                            System.currentTimeMillis()
                    );
                }
                return null;
            }
        });

// 输出到 ClickHouse
        DataStream<String> alertGiftStringStream = alertStream.map(new MapFunction<AlertGiftSender, String>() {
            @Override
            public String map(AlertGiftSender value) throws Exception {
                return value.getUserId() + "," +
                        value.getWindowStartTime() + "," +
                        value.getWindowEndTime() + "," +
                        value.getSendMoney() + "," +
                        value.getTs();
            }
        });

        JdbcUtil.sinkToClickhouseUpsert(alertGiftStringStream, "insert into real_time_tiktok_report.tiktok_alert_gift values(?,?,?,?,?)");





        //机器人发言检测
        WindowedStream<TikTokBean, String, TimeWindow> messageWindowStream = watermarks
                // 1. 过滤发言事件
                .filter(event -> "message".equals(event.getEventType()))
                // 2. 按用户ID分组
                .keyBy(TikTokBean::getUserId)
                // 3. 定义滚动计数窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(1)));
        DataStream<AlertMessageUser> messageStream = messageWindowStream.aggregate(new MessageCountAggregator(), new MessageAlertProcessFunction());

        DataStream<String> streamOperator = messageStream.map(new MapFunction<AlertMessageUser, String>() {
            @Override
            public String map(AlertMessageUser value) throws Exception {
                return value.getUserId() + "," + value.getWindowStartTime() + "," + value.getWindowEndTime() + "," + value.getMessageCount() + "," + value.getTs();
            }
        });


        JdbcUtil.sinkToClickhouseUpsert(streamOperator, "insert into real_time_tiktok_report.tiktok_alert_message values(?,?,?,?,?)");

        env.execute("KafkaDataEtlJob");
    }

    private static DataStream<String> etlNoJsonData(DataStream<String> kafkaDataStream) {

        OutputTag<String> err = new OutputTag<String>("err") {
        };

        // DataStream etl 过滤脏数据
        DataStream<String> etlDataStream = kafkaDataStream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value,
                                       ProcessFunction<String, String>.Context ctx,
                                       Collector<String> out) throws Exception {
                try {

                    JSON.parseObject(value);
                    out.collect(value);
                } catch (Exception e) {
                    ctx.output(err, value);
                }
            }
        });
        return etlDataStream;
    }


}
